package com.rxmvp.baselibrary.rxbus;


import com.trello.rxlifecycle2.LifecycleProvider;

import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 *  RxJava实现的事件总线
 *
 * 作者： huangRZ on 2017/1/21. 15:04
 * 邮箱：917647409@qq.com
 */
public class RxBus {

    private static RxBus rxBus;
    private final Subject<Events<?>> _bus = PublishSubject.<Events<?>>create().toSerialized();

    private RxBus(){}

    public static RxBus getInstance(){
        if (rxBus == null){
            synchronized (RxBus.class){
                if (rxBus == null){
                    rxBus = new RxBus();
                }
            }
        }
        return rxBus;
    }

    public void send(Events<?> o) {
        _bus.onNext(o);
    }

    public Observable<Events<?>> toObservable() {
        return _bus;
    }

    public void send(@Events.EventCode int code, Object content){
        Events<Object> event = new Events<>();
        event.setCode(code);
        event.setContent(content);
        send(event);
    }

    public static <E> SubscriberBuilder<E> with(LifecycleProvider<E> provider){
        return new SubscriberBuilder<>(provider);
    }


    /**
     *  E：ActivityEvent  或者  FragmentEvent
     */
    public static class SubscriberBuilder<T>{

        private LifecycleProvider<T> mLifecycleProvider;
        private T mEndEvent;
        private int event;
        private Consumer<? super Events<?>> onNext;
        private Consumer<Throwable> onError;

        public SubscriberBuilder(LifecycleProvider<T> provider) {
            this.mLifecycleProvider = provider;
        }

        public SubscriberBuilder<T> setEvent(@Events.EventCode int event){
            this.event = event;
            return this;
        }

        public SubscriberBuilder<T> setLifecycleEvent(T event){
            this.mEndEvent = event;
            return this;
        }

        public SubscriberBuilder<T> onNext(Consumer<? super Events<?>> action){
            this.onNext = action;
            return this;
        }

        public SubscriberBuilder<T> onError(Consumer<Throwable> action){
            this.onError = action;
            return this;
        }

        public void create(){
            _create();
        }

        public Disposable _create(){
            if (mLifecycleProvider !=null){
                return RxBus.getInstance().toObservable()
                        .compose(mEndEvent == null ? mLifecycleProvider.<Events<?>>bindToLifecycle() : mLifecycleProvider.<Events<?>>bindUntilEvent(mEndEvent)) // 绑定生命周期
                        .filter(new Predicate<Events<?>>() {
                            @Override
                            public boolean test(Events<?> events) throws Exception {
                                return events.getCode() == event;
                            }
                        })   //过滤 根据code判断返回事件
                        .subscribe(onNext, onError == null ? new Consumer<Throwable>() {
                            @Override
                            public void accept(Throwable throwable) throws Exception {
                                throwable.printStackTrace();
                            }
                        } : onError);
            }
            return null;
        }
    }
}
